// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include <algorithm>

#include <string>
#include <map>

#include "port/stack_trace.h"
#include "db/db_test_util.h"
#include "db/column_family.h"
#include "db/db_impl.h"
#include "db/flush_job.h"
#include "pure_mem/memoryblock/memory_arena/memory_arena.h"
#include "pure_mem/std_out_logger.h"
#include "rocksdb/cache.h"
#include "rocksdb/env.h"
#include "util/file_reader_writer.h"
#include "util/testharness.h"
#include "util/testutil.h"

namespace rocksdb {

const size_t kApproximateSize = 10 << 20;
const size_t kSst_size = 1 << 15;

class FlushL0CacheTest : public DBTestBase {
 public:
  FlushL0CacheTest() : DBTestBase("/db_basic_test"), env_options_(last_options_) {
    last_options_.info_log.reset(new StdOutLogger(InfoLogLevel::DEBUG_LEVEL));
    FlushL0CacheTryReopen();
  }

  Status FlushL0CacheTryReopen() {
    Close();
    return DB::Open(last_options_, dbname_, &db_);
  }

  EnvOptions env_options_;

  static Slice genMVCCKey1(const char* key) {
    uint32_t keySize = strlen(key) + 1;
    char* ret = new char[keySize];
    memset(ret, '\0', keySize);
    memcpy(ret, key, strlen(key));
    return Slice(ret, keySize);
  };

  static Slice CreateInternalKeySlice(SequenceNumber seq);
  static void InsertData(MemArena *memory_arena, int size);
  static std::map<std::string, std::string> InsertDataAndDelete(MemArena *memory_arena, int data_size,
                           int del_nums, int next_key_nums);
  static void InsertDataAndRangeDel(MemArena *memory_arena, int data_size,
                             int range_del_nums, int range_del_size,
                             int next_key_nums, SequenceNumber range_del_seq);

};

Slice FlushL0CacheTest::CreateInternalKeySlice(SequenceNumber seq) {
  std::string a = std::to_string(seq) + "_test";
  Slice key = Slice(genMVCCKey1(a.data()));

  auto key_size = static_cast<uint32_t>(key.size());
  uint32_t internal_key_size = key_size + 8;

  char *buf = new char[internal_key_size];
  char *p = buf;
  memcpy(p, key.data(), key_size);
  p += key_size;
  uint64_t packed = PackSequenceAndType(seq, kTypeValue);
  EncodeFixed64(p, packed);
  return {buf, internal_key_size};
}

void FlushL0CacheTest::InsertData(MemArena *memory_arena, int size = 10) {
  char tmp[1000] = {'a'};
  Slice value(tmp, 1000);
  SequenceNumber s = 100000;
  for (int i = 0; i < size; i++) {
    s++;
    std::string a = std::to_string(s) + "_test";
    Slice key = Slice(genMVCCKey1(a.data()));

    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;

    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(s, kTypeValue);
    EncodeFixed64(p, packed);

    Slice key_slice(buf, internal_key_size);

    // 内存存储区插入数据
    memory_arena->KVDataInsert(key_slice, value);
  }
}

std::map<std::string, std::string> FlushL0CacheTest::InsertDataAndDelete(MemArena *memory_arena, int data_size = 10,
                         int del_nums = 1, int next_key_nums = 5) {
  char tmp[1000] = {'a'};
  Slice value(tmp, 1000);
  SequenceNumber data_seq = 100000;
  SequenceNumber del_seq = 100000;
  std::map<std::string, std::string> kv_map;

  // wangzekun01: 插入数据
  for (int i = 0; i < data_size; i++) {
    ++data_seq;
    std::string a = std::to_string(data_seq) + "_test";
    Slice key = Slice(genMVCCKey1(a.data()));

    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;

    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(data_seq, kTypeValue);
    EncodeFixed64(p, packed);

    Slice key_slice(buf, internal_key_size);

    // 内存存储区插入数据
    memory_arena->KVDataInsert(key_slice, value);
    kv_map[key_slice.ToString()] = value.ToString();
  }

  // wangzekun01: 插入删除数据
  for (int i = 0; i < del_nums; i++) {
    std::string start_key = std::to_string(++del_seq) + "_test";

    ++data_seq;
    Slice key = Slice(genMVCCKey1(start_key.data()));
    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;
    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(data_seq, kTypeDeletion);
    EncodeFixed64(p, packed);
    Slice key_slice(buf, internal_key_size);

    Slice value_slice;
    // 内存存储区插入范围删除数据
    memory_arena->KVDataInsert(key_slice, value_slice);
    kv_map[key_slice.ToString()] = value_slice.ToString();

    del_seq += next_key_nums - 1;
  }

  // 向内存存储区插入一条已删除数据
  {
    ++data_seq;
    int tmp_seq = next_key_nums * 2 + 1 + 100000;
    std::string a = std::to_string(tmp_seq) + "_test";
    Slice key = Slice(genMVCCKey1(a.data()));

    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;

    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(data_seq, kTypeValue);
    EncodeFixed64(p, packed);

    Slice key_slice(buf, internal_key_size);

    // 内存存储区插入数据
    memory_arena->KVDataInsert(key_slice, value);
    kv_map[key_slice.ToString()] = value.ToString();
  }
  return kv_map;
}

void FlushL0CacheTest::InsertDataAndRangeDel(MemArena *memory_arena, int data_size = 10,
                           int range_del_nums = 1, int range_del_size = 2,
                           int next_key_nums = 5, SequenceNumber range_del_seq = 100000) {

  char tmp[1000] = {'a'};
  Slice value(tmp, 1000);
  SequenceNumber data_seq = 100000;

  // wangzekun01: 插入数据
  for (int i = 0; i < data_size; i++) {
    data_seq++;
    std::string a = std::to_string(data_seq) + "_test";
    Slice key = Slice(genMVCCKey1(a.data()));

    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;

    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(data_seq, kTypeValue);
    EncodeFixed64(p, packed);

    Slice key_slice(buf, internal_key_size);

    // 内存存储区插入数据
    memory_arena->KVDataInsert(key_slice, value);
  }

  // wangzekun01: 插入范围删除数据
  for (int i = 0; i < range_del_nums; i++) {
    // wangzekun01: 获取范围删除的start key和end key。
    std::string start_key = std::to_string(++range_del_seq) + "_test";
    std::string end_key = std::to_string(range_del_seq + range_del_size) + "_test";

    ++data_seq;
    Slice key = Slice(genMVCCKey1(start_key.data()));
    auto key_size = static_cast<uint32_t>(key.size());
    uint32_t internal_key_size = key_size + 8;
    char buf[internal_key_size];
    char *p = buf;
    memcpy(p, key.data(), key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(data_seq, kTypeRangeDeletion);
    EncodeFixed64(p, packed);
    Slice key_slice(buf, internal_key_size);

    Slice value_slice(genMVCCKey1(end_key.data()));

    // 内存存储区插入范围删除数据
    memory_arena->RangeDeletionInsert(key_slice, value_slice);

    range_del_seq += next_key_nums - 1;
  }
}

TEST_F(FlushL0CacheTest, BasicFlush) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertData(memory_arena);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;
  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 1);
  ASSERT_EQ(genMVCCKey1((std::to_string(100001) + "_test").data()), l1_files[0]->smallest.user_key());
  ASSERT_EQ(genMVCCKey1((std::to_string(100010) + "_test").data()), l1_files[0]->largest.user_key());

  // SST数据插入检查
  ReadRangeDelAggregator range_del_aggregator(&default_cfd->internal_comparator(), kMaxSequenceNumber);
  std::unique_ptr<InternalIterator> it(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[0],
      &range_del_aggregator /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, true /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));
  int seq = 100000;
  for (it->SeekToFirst(); it->Valid(); it->Next()) {
    ++seq;
    ASSERT_EQ(it->user_key(), genMVCCKey1((std::to_string(seq) + "_test").data()));
    ASSERT_EQ(it->value().ToString()[0], 'a');
  }
}

TEST_F(FlushL0CacheTest, SstSizeControl) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertData(memory_arena, 100);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kSst_size, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 4);
}

TEST_F(FlushL0CacheTest, InsertAndDeleteFlush) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  std::map<std::string, std::string> TEST_result = InsertDataAndDelete(memory_arena, 10, 3, 3);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 1);

  // SST数据插入检查
  ReadRangeDelAggregator range_del_aggregator(&default_cfd->internal_comparator(), kMaxSequenceNumber);
  std::unique_ptr<InternalIterator> it(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[0],
      &range_del_aggregator /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, true /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));
  for (it->SeekToFirst(); it->Valid(); it->Next()) {
    ASSERT_EQ(it->value().ToString(), TEST_result[it->key().ToString()]);
  }
}

TEST_F(FlushL0CacheTest, LargeMemoryArenaFlush) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertData(memory_arena, 40000);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // 文件个数检查
  std::vector<LiveFileMetaData> live_files_meta;
  dbfull()->GetLiveFilesMetaData(&live_files_meta);
  ASSERT_EQ(live_files_meta.size(), 4);

  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  int num_entries = 0;
  for (const auto &l1_file : l1_files) {
    num_entries += l1_file->num_entries;
  }
  ASSERT_EQ(l1_files.size(), 4);
  ASSERT_EQ(num_entries, 40000);
}

TEST_F(FlushL0CacheTest, RangeDeletionFlush) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertDataAndRangeDel(memory_arena, 10, 3, 3);
  std::vector<std::string> TEST_result = {genMVCCKey1("100004_test\0").ToString(),
                                    genMVCCKey1("100005_test\0").ToString(),
                                    genMVCCKey1("100009_test\0").ToString(),
                                    genMVCCKey1("100010_test\0").ToString()};
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 1);
  ASSERT_EQ(l1_files[0]->smallest.user_key(), genMVCCKey1("100001_test\0"));
  ASSERT_EQ(l1_files[0]->largest.user_key(), genMVCCKey1("100014_test\0"));

  // SST数据插入检查
  std::unique_ptr<InternalIterator> it(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[0],
      nullptr /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, false /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));

  int test_data_sum = 0;
  for (it->SeekToFirst(); it->Valid(); it->Next(), ++test_data_sum) {
    ASSERT_EQ(TEST_result[test_data_sum], it->user_key().ToString());
  }
  ASSERT_EQ(test_data_sum, 4);
}

TEST_F(FlushL0CacheTest, RangeDelCutTablesFlush) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertDataAndRangeDel(memory_arena, 50000, 3,
                        10000, 20000, 100000);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;

  snapshot_seqs = {125000};
  earliest_write_conflict_snapshot = 125000;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 4);
  int num_entries = 0;
  for (const auto &l1_file : l1_files) {
    num_entries += l1_file->num_entries;
  }
  ASSERT_EQ(num_entries, 35004);
}

TEST_F(FlushL0CacheTest, snapshotsTable) {
  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertDataAndRangeDel(memory_arena, 10, 1,
                        6, 0, 100002);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  std::vector<Slice> empty_boundaries;
  snapshot_seqs = {100005};
  earliest_write_conflict_snapshot = 100005;

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, 3000, empty_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);

  // SST数据插入检查
  ReadRangeDelAggregator range_del_aggregator(&default_cfd->internal_comparator(), kMaxSequenceNumber);
  std::unique_ptr<InternalIterator> it(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[0],
      &range_del_aggregator /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, true /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));

  ReadRangeDelAggregator range_del_aggregator2(&default_cfd->internal_comparator(), kMaxSequenceNumber);
  std::unique_ptr<InternalIterator> it2(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[1],
      &range_del_aggregator2 /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, true /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));

  ASSERT_EQ(l1_files[0]->num_entries, 4);
  ASSERT_EQ(l1_files[1]->num_entries, 4);
  ASSERT_EQ(l1_files[0]->num_deletions, 1);
  ASSERT_EQ(l1_files[1]->num_deletions, 1);
}

TEST_F(FlushL0CacheTest, BoundaiesCutTable) {

  // 为默认列族添加内存存储区和数据
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  ColumnFamilyData *default_cfd = cfh->cfd();
  auto memory_arena = new MemArena(default_cfd->internal_comparator(), *default_cfd->ioptions());
  InsertData(memory_arena);
  memory_arena->Ref();
  default_cfd->SetMemoryarena(memory_arena);

  // 创建其他所需参数
  JobContext job_context(0);
  MutableCFOptions mutable_cf_options = *default_cfd->GetLatestMutableCFOptions();
  std::vector<SequenceNumber> snapshot_seqs;
  SequenceNumber earliest_write_conflict_snapshot;
  SnapshotChecker *snapshot_checker;
  VersionEdit edit;;
  std::vector<FileMetaData> metas;
  dbfull()->mutex()->Lock();
  dbfull()->GetSnapshotContext(&job_context, &snapshot_seqs,
                               &earliest_write_conflict_snapshot, &snapshot_checker);
  // 添加边界值
  Slice key_slice1 = CreateInternalKeySlice(100003);
  Slice key_slice2 = CreateInternalKeySlice(100006);
  std::vector<Slice> key_boundaries = {key_slice1, key_slice2};
  std::vector<std::pair<std::string, std::string > > TEST_result = {
      {genMVCCKey1("100001_test\0").ToString(), genMVCCKey1("100003_test\0").ToString()},
      {genMVCCKey1("100004_test\0").ToString(), genMVCCKey1("100006_test\0").ToString()},
      {genMVCCKey1("100007_test\0").ToString(), genMVCCKey1("100010_test\0").ToString()},
  };

  // 写入数据
  Status s = dbfull()->FlushL0CacheToSST(default_cfd, mutable_cf_options, &job_context,
                                         snapshot_seqs, earliest_write_conflict_snapshot,
                                         snapshot_checker, Env::HIGH, kApproximateSize, key_boundaries,
                                         edit, metas);
  s = default_cfd->current()->version_set()->LogAndApply(default_cfd, *default_cfd->GetLatestMutableCFOptions(),
                                                         &edit, dbfull()->mutex());
  dbfull()->mutex()->Unlock();

  // Manifest检查
  const auto &l1_files = default_cfd->current()->storage_info()->LevelFiles(1);
  ASSERT_EQ(l1_files.size(), 3);
  int threshold = l1_files.size();
  for (int i = 0; i < threshold; ++i) {
    ASSERT_EQ(TEST_result[i].first, l1_files[i]->smallest.user_key().ToString());
    ASSERT_EQ(TEST_result[i].second, l1_files[i]->largest.user_key().ToString());
  }

  // SST数据插入检查
  ReadRangeDelAggregator range_del_aggregator(&default_cfd->internal_comparator(), kMaxSequenceNumber);
  std::unique_ptr<InternalIterator> it(default_cfd->table_cache()->NewIterator(
      ReadOptions(), env_options_, default_cfd->internal_comparator(), *l1_files[0],
      &range_del_aggregator /* range_del_agg */, mutable_cf_options.prefix_extractor.get(),
      nullptr, nullptr, true /* for_compaction */, nullptr /* arena */,
      false /* skip_filter */, 1));
  int seq = 100000;
  for (it->SeekToFirst(); it->Valid(); it->Next()) {
    ++seq;
    ASSERT_EQ(it->user_key(), genMVCCKey1((std::to_string(seq) + "_test").data()));
    ASSERT_EQ(it->value().ToString()[0], 'a');
  }
}

TEST_F(FlushL0CacheTest, L0CacheInsertFlushTest) {
  Options options = CurrentOptions();
  options.rocksdb_l0_cache_flush_l1 = true;
  options.level0_file_num_compaction_trigger = 2;
  Reopen(options);
  for (int i = 0; i < 50; i++) {
    std::string key1 = "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu"));
  }
  ASSERT_EQ("pikachu", Get(genMVCCKey1("a11").ToString()));
  FlushOptions flush_options;
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  MemArena *memArena = cfh->cfd()->memoryarena();
  Slice key_find("a15");
  int len = (int) key_find.size() + 8 + 1;
  uint32_t keySize = VarintLength(len) + len;
  char find[keySize];
  memset(find, '\0', keySize);
  memcpy(find, key_find.data(), key_find.size());
  memset(find + keySize - 8, 255, 8);
  Slice key(find, keySize);
  ASSERT_EQ(memArena->Seek(key)->Valid(), true);
  for (int i = 0; i < 50; i++) {
    std::string key2 = "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key2.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
}

TEST_F(FlushL0CacheTest, L0CacheInsertFlushGetTest) {
  Options options = CurrentOptions();
  options.rocksdb_l0_cache_flush_l1 = true;
  options.level0_file_num_compaction_trigger = 2;
  ReadOptions readOptions;
  Reopen(options);
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  for (int i = 0; i < 50; i++) {
    std::string key1 = "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu" + ToString(i)));
  }
  Slice key_find("a15");
  PinnableSlice value_get;
  Slice get_key = genMVCCKey1(key_find.data());
  // Get from memtable
  auto s = dbfull()->GetImpl(readOptions, cfh, get_key,  &value_get);
  ASSERT_OK(s.OK());
  ASSERT_EQ("pikachu15", *value_get.GetSelf());

  FlushOptions flush_options;
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  //MemoryArena *memArena = cfh->cfd()->memoryarena();

  PinnableSlice value_get1;
  // Get from memArena
  s = dbfull()->GetImpl(readOptions, cfh, get_key,  &value_get1);
  ASSERT_TRUE(s.ok());
  ASSERT_EQ("pikachu15", *value_get1.GetSelf());

  for (int i = 0; i < 50; i++) {
    std::string key2 = "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key2.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
  PinnableSlice value_get2;
  // Get from L1
  s = dbfull()->GetImpl(readOptions, cfh, get_key,  &value_get2);
  ASSERT_TRUE(s.ok());
  ASSERT_EQ("pikachu", value_get2.ToString());
}

TEST_F(FlushL0CacheTest, L0CacheInsertFlushMultiTest) {
  Options options = CurrentOptions();
  options.rocksdb_l0_cache_flush_l1 = true;
  options.level0_file_num_compaction_trigger = 2;
  Reopen(options);
  for (int i = 0; i < 50; i++) {
    std::string key1 = "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu"));
  }
  ASSERT_EQ("pikachu", Get(genMVCCKey1("a11").ToString()));
  FlushOptions flush_options;
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  for (int i = 0; i < 50; i++) {
    std::string key2 = "b" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key2.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
  for (int i = 0; i < 50; i++) {
    std::string key1 = "c" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  for (int i = 0; i < 50; i++) {
    std::string key2 = "d" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key2.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  for (int i = 0; i < 50; i++) {
    std::string key1 = "e" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  for (int i = 0; i < 50; i++) {
    std::string key2 = "f" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key2.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
}

TEST_F(FlushL0CacheTest, L0CacheRangeDeletionFlushGetTest) {
  Options options = CurrentOptions();
  options.rocksdb_l0_cache_flush_l1 = true;
  options.level0_file_num_compaction_trigger = 2;
  ReadOptions readOptions;
  Reopen(options);
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl *>(dbfull()->DefaultColumnFamily());
  for (int i = 0; i < 50; i++){
    std::string key =  "a" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->DeleteRange(WriteOptions(), dbfull()->DefaultColumnFamily(),
                                        genMVCCKey1("a10"), genMVCCKey1("a20")));
  ASSERT_EQ("NOT_FOUND", Get(genMVCCKey1("a11").ToString()));
  Slice key_find("a11");
  PinnableSlice value_get;
  Slice get_key = genMVCCKey1(key_find.data());
  // Get from memtable
  auto s = dbfull()->GetImpl(readOptions, cfh, get_key,  &value_get);
  ASSERT_TRUE(s.IsNotFound());
  FlushOptions flush_options;
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
  // Get from memArena
  s = dbfull()->GetImpl(readOptions, cfh, get_key,  &value_get);
  ASSERT_TRUE(s.IsNotFound());

  for (int i = 0; i < 50; i++){
    std::string key1 =  "b" + ToString(i);
    ASSERT_OK(Put(genMVCCKey1(key1.data()), "pikachu"));
  }
  ASSERT_OK(dbfull()->Flush(flush_options));
  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
}

}  // namespace rocksdb

int main(int argc, char **argv) {
  rocksdb::port::InstallStackTraceHandler();
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}